package com.ruoyi.mqtt.client;

import com.alibaba.fastjson.JSON;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.data.es.domain.SensorData;
import com.ruoyi.data.tdengine.domain.Sensor;
import com.ruoyi.framework.manager.AsyncManager;
import com.ruoyi.mqtt.factory.AsyncFactory;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @ClassName MqttPushCallback
 * @Author Iven
 * @Description TODO
 * @Date 2020/12/6 23:23
 */
public class MqttPushCallback implements MqttCallback {

    private static final Logger log = LoggerFactory.getLogger(MqttPushCallback.class);

    @Override
    public void connectionLost(Throwable cause) {
        log.info("断开连接，建议重连" + this);
        MqttPushClient client = MqttPushClient.getInstance();
        while (true) {
            try {
                //判断拦截状态，这里注意一下，如果没有这个判断，是非常坑的
                if (!client.isConnected()) {
                    System.out.println("***** client to connect *****");
                    client.connect();
                }
                if (client.isConnected()) {//连接成功，跳出连接
                    System.out.println("***** connect success *****");
                    break;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        client.connect();
        client.subscribe("/iot/data/#");
        //断开连接，建议重连
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        //log.info(token.isComplete() + "");
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("Topic: " + topic);
        String msg = new String(message.getPayload());
        log.info("Message: " + msg);
        msg = msg.substring(0,msg.lastIndexOf("]")+1);
        if (topic.startsWith("/iot/data/")){
            String equipmentId = StringUtils.substringAfter(topic,"/iot/data/");
            List<SensorData> sensorDatas = JSON.parseArray(msg,SensorData.class);
            for(SensorData data: sensorDatas)
            {
                data.setAddTime(new Date());
                data.setEquipmentCode(equipmentId);
            }
            List<Sensor> sensors = sensorDatas.stream().map(sensorData -> {
                Sensor sensor = new Sensor();
                sensor.setSensorCode(sensorData.getCode());
                sensor.setEquipmentCode(sensorData.getEquipmentCode());
                sensor.setValue(sensorData.getValue());
                sensor.setTs(new Timestamp(System.currentTimeMillis()));
                return  sensor;
            }).collect(Collectors.toList());
            //AsyncManager.me().execute(AsyncFactory.saveData2ElasticSearch(SensorDatas));
            AsyncManager.me().execute(AsyncFactory.saveData2Tdengine(sensors));
        }

    }

}
